package com.chen.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.ObjUtil;
import com.alibaba.dashscope.aigc.generation.GenerationResult;
import com.alibaba.dashscope.common.Message;
import com.alibaba.dashscope.common.Role;
import com.chen.bean.constants.WxConstants;
import com.chen.bean.entity.DiaLogEntity;
import com.chen.bean.entity.MessageEntity;
import com.chen.bean.vo.DialogVo;
import com.chen.bean.vo.MessageRequest;
import com.chen.exception.BusinessException;
import com.chen.manager.QwenManager;
import com.chen.service.ChatService;
import com.chen.service.DialogService;
import com.chen.service.MessageService;
import com.chen.util.CacheUtils;
import com.chen.util.DiaLogNumberGenerator;
import com.chen.util.JsonMapper;
import com.fasterxml.jackson.core.type.TypeReference;
import io.reactivex.Flowable;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

@Service
@Slf4j
public class ChatServiceImpl implements ChatService {

    @Autowired
    private QwenManager qwenManager;

    @Autowired
    private CacheUtils cacheUtils;

    @Autowired
    private MessageService messageService;

    @Autowired
    private DialogService dialogService;

    @Autowired
    private DiaLogNumberGenerator numberGenerator;

    @Override
    public Flux<String> chat(MessageRequest request,Integer userId) throws BusinessException {
        String dialogNumber = request.getDialogNumber();
        List<MessageEntity> historyEntityList = new ArrayList<>();
        //请求通义千问消息体
        List<Message> messageList = new ArrayList<Message>();
        Message userMsg = Message.builder().role(Role.USER.getValue()).content(request.getMessage()).build();
        //历史消息缓存拼接用户id
        String messageSuffixCacheKey = WxConstants.WX_HISTORY_MESSAGE_CACHE + userId;

        String json = JsonMapper.toJson(cacheUtils.getCacheList(messageSuffixCacheKey+ ":" + dialogNumber));
        historyEntityList = JsonMapper.formJson(json, new TypeReference<List<MessageEntity>>() {
        });
        //缓存为空查询数据库
        if (CollectionUtil.isEmpty(historyEntityList)){
            historyEntityList =  messageService.queryByDialogNumber(dialogNumber);
        }
        List<Message> historyMessageList = messageService.buildMessageList(historyEntityList);
        messageList.addAll(historyMessageList);

        //本次用户本次所发消息
        MessageEntity messageEntity = messageService.buildMessageEntity(dialogNumber, userMsg.getRole(), userMsg.getContent());
        historyEntityList.add(messageEntity);
        messageList.add(userMsg);
        //调用通义千问
        Flowable<GenerationResult> resultFlowable = qwenManager.qwenStreamCall(messageList);
        // 将 Flowable 转换为 Flux
        Flux<GenerationResult> generationResultFlux = Flux.from(resultFlowable);
        StringBuffer sb = new StringBuffer();
        //本次消息输出总长度
        AtomicInteger outputTokens = new AtomicInteger(0);
        //本次系统所回消息,使用atom保证线程安全
        AtomicReference<MessageEntity> assistantEntity = new AtomicReference<>();
        assistantEntity.set(new MessageEntity());
        List<MessageEntity> finalHistoryEntityList = historyEntityList;
        return generationResultFlux
                .map(result -> {
                    Message message = result.getOutput().getChoices().get(0).getMessage();
                    String content = message.getContent();
                    outputTokens.addAndGet(result.getUsage().getOutputTokens());
                    assistantEntity.get().setRole(message.getRole());
                    assistantEntity.get().setInputTokens(result.getUsage().getInputTokens());
                    sb.append(content);
                    return content;
                }).doFinally(signalType -> {
                    log.info("对话结束,signalType结果："+signalType);
                    assistantEntity.updateAndGet(entity ->{
                        entity.setContent(sb.toString());
                        entity.setOutputTokens(outputTokens.get());
                        Integer totalTokens =entity.getInputTokens() + entity.getOutputTokens();
                        entity.setTotalTokens(totalTokens);
                        entity.setDialogNumber(dialogNumber);
                        entity.setCreateTime(new Date());
                        entity.setUpdateTime(new Date());
                        return entity;
                    });
                    finalHistoryEntityList.add(assistantEntity.get());
                    //消息内容暂存redis,异步执行
                    messageService.saveMessageToRedis(messageSuffixCacheKey + ":" + dialogNumber,finalHistoryEntityList);
                });
    }

    /**
     * redis同步mysql
     * dialogNumber举例 cache:historyMessage:15:2024111414000011
     */
    @Override
    @Async("syncMessage")
    public void syncMessage(String messageKey) {
        String dialogNumber = StringUtils.substringAfterLast(messageKey, ":");
        //对话记录存入数据库，并清除redis缓存
        String json = JsonMapper.toJson(cacheUtils.getCacheList(messageKey));
        List<MessageEntity> cacheList = JsonMapper.formJson(json, new TypeReference<List<MessageEntity>>() {
        });
        if (CollectionUtil.isEmpty(cacheList)) return;
        //存消息记录 去重已经存入的，id不为空说明已入库
        List<MessageEntity> distinct = cacheList.stream().filter(entity -> ObjUtil.isNull(entity.getId())).toList();
        if (CollectionUtil.isNotEmpty(distinct)){
            messageService.batchSave(distinct);
            DiaLogEntity entity = dialogService.queryEntityByDialogNumber(dialogNumber);
            //dialogName赋值一次就可以了
            if (ObjUtil.isNotNull(entity) && StringUtils.isBlank(entity.getDialogName())){
                String content = cacheList.get(0).getContent();
                // 取前十位字符防止超出数据库长度
                String dialogName = content.length() > 10 ? content.substring(0, 10) : content;
                entity.setDialogName(dialogName);
                entity.setUpdateTime(new Date());
                dialogService.update(entity);
            }
            log.info("messageKey = " + messageKey);
        }
        //清理缓存
        cacheUtils.deleteObject(messageKey);
    }

    @Override
    public String generatorDialogNumber(Integer userId) {
        return numberGenerator.generateDialogNumber();
    }
}
